package com.example.mq.mqserver.core;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.VirtualHost;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * 消费消息的核心逻辑
 */
public class ConsumerManager {

    // 持有上层的 VirtualHost 对象的引用，操作数据
    private VirtualHost parent;

    // 指定一个线程池，负责去执行具体的回调任务
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);

    // 存放令牌的队列
    private BlockingQueue<String> tokenQueue = new LinkedBlockingDeque<>();

    // 扫描线程
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost p){
        parent = p;

        scannerThread = new Thread(() -> {
           while (true){
               try{
                   // 1. 拿到令牌
                   String queueName = tokenQueue.take();
                   // 2. 根据令牌，找到队列
                   MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                   if(null == queue){
                       throw new MqException("[ConsumerManager] 取令牌后发现，该队列名不存在！ queueName = " + queueName);
                   }
                   // 3. 从这个队列中消费这个消息
                   synchronized (queue){
                       consumeMessage(queue);
                   }
               } catch (InterruptedException | MqException e) {
                   e.printStackTrace();
               }
           }
        });
        // 改成后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    // 这个方法调用时机是发送消息的时候
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    // 将 ConsumerEnv 对象添加到指定队列中
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        // 找到对应的队列
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if(null == queue){
            throw new MqException("[ConsumerManager] 队列不存在！ queueName = " + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        synchronized (queue){
            queue.addConsumerEnv(consumerEnv);
            // 如果当前队列中已经有了一些消息了，需要立即就消费掉
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                // 调用一次就消费一次消息
                consumeMessage(queue);
            }
        }
    }

    private void consumeMessage(MSGQueue queue) {

        // 1. 按照轮询的方式，找个消费者
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if(null == luckyDog){
            // 当前队列没有消费者
            return;
        }
        // 2. 从队列中取出一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(null == message){
            // 当前队列还没有消息
            return;
        }
        // 3. 把消息带入到消费者的回调方法中，丢给线程池执行
        workerPool.submit(() -> {
            try{
                // 1. 把消息放到待确认的集合中，这个操作一定在执行回调钱
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                // 2. 真正执行回调操作
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(), message.getBasicProperties(), message.getBody());
                // 3. 如果当前是 "自动应答"，就可以直接将该消息删除
                //    如果当前是 "手动应答"，，则先不处理，交给后续消费者调用 basicAck 方法来处理
                if(luckyDog.isAutoAck()){
                    // 1) 删除硬盘上的消息
                    if(message.getDeliverMode() == 2){
                        parent.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    // 2) 删除上面的待确认集合中的消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    // 3) 删除内存中消息中心中的消息
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    System.out.println("[ConsumerManager] 消息被成功消费！ queueName = " + queue.getName());
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        });
    }
}
